package com.gate.ws.spring.rocketmq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.gate.ws.spring.cache.LocalChannelCache;
import com.gate.ws.spring.utils.JsonUtils;
import com.nami.basic.constants.CommonConst;
import com.nami.gate.api.request.message.DownMsgRequest;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

import static com.nami.basic.constants.DownTopicConst.FLUSH_ONE_TOPIC;


@Slf4j
@Component
@RocketMQMessageListener(topic = FLUSH_ONE_TOPIC, consumerGroup = "${rocketmq.consumer.flushOneConsumerGroup}", selectorExpression = "${nami.mqTags}", consumeMode = ConsumeMode.CONCURRENTLY)
public class DownOneMsgConsumer implements RocketMQListener<String> {

    @Override
    public void onMessage(String msg) {
        try {
            DownMsgRequest request = JsonUtils.parseObject(msg, DownMsgRequest.class);
            TimeInterval timer = DateUtil.timer();
            if ((System.currentTimeMillis() - request.getCurrTime()) > CommonConst.DELTA) {
                log.warn("[flushOne] 过期信息不予消费。identityNo={}，messageId={}", request.getIdentityNo(), request.getMessageId());
                return;
            }
            log.info("[flushOne] 发送消息 identityNo={} msgId={}", request.getIdentityNo(), request.getMessageId());
            Channel channel = LocalChannelCache.get(request.getIdentityNo());
            if (channel == null) {
                // 如果为空，暂时为删除
//                LocalChannelCache.removeAll(request.getIdentityNo());
                log.error("[flushOne] 发送失败，通道为空 identityNo={}", request.getIdentityNo());
                return;
            }
            channel.writeAndFlush(request.getPacket());
            if (timer.interval() > 500) {
                log.warn("[flushOne]【{}】发送消息成功 use:{}ms", request.getIdentityNo(), timer.interval());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
